package com.cumt.tool.kafka;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by caicf on 2017/1/12.
 */
public class KafkaConsumerUtil {

    private static final Logger log = LoggerFactory.getLogger(KafkaConsumerUtil.class);

    private static final ReentrantLock lock = new ReentrantLock();

    private static KafkaConsumerUtil instance;

    private static ExecutorService executor = Executors.newCachedThreadPool();

    private KafkaConsumerUtil() {

    }

    public static KafkaConsumerUtil getInstance() {
        if (instance == null) {
            lock.lock();
            if (instance == null) {
                instance = new KafkaConsumerUtil();
            }
            lock.unlock();
        }
        return instance;
    }

    public void consumer(int consumerNum, String groupId, List<String> topics) {
//        int numConsumers = 10;
//        String groupId = "1";
//        List<String> topics = Arrays.asList("cumt2");

        final List<ConsumerLoop> consumers = new ArrayList<>();
        for (int i = 0; i < consumerNum; i++) {
            ConsumerLoop consumer = new ConsumerLoop(i, groupId, topics);
            consumers.add(consumer);
            executor.submit(consumer);
        }

        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                for (ConsumerLoop consumer : consumers) {
                    consumer.shutdown();
                }
                executor.shutdown();
                try {
                    executor.awaitTermination(5000, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }
}
